package com.atguigu.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.atguigu.api.SensorReading
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

/**
 * @ClassName JDBCSinkTest
 * @Description
 * @Author Mr Yang
 * @Date 2020/9/1 22:04
 * @Version 1.0
 */
object JDBCSinkTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //source
    val inputStream = env.readTextFile("F:\\work\\FlinkTutorial\\src\\main\\resources\\sensor.txt")
    //transform
    val dataStream = inputStream.map(
      data => {
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      }
    )
    //sink
    dataStream.addSink( new MyJDBCSink() )

    env.execute("JDBC sink test")
  }
}

class MyJDBCSink() extends RichSinkFunction[SensorReading]{
  //定义sql连接，预编译器
  var conn: Connection = _
  var insertStat: PreparedStatement = _
  var updateStat: PreparedStatement = _

  //初始化，创建连接和预编译语句
  override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
    insertStat = conn.prepareStatement("INSERT INTO temperatures (sensor_id, temperature) VALUES (?,?)")
    updateStat = conn.prepareStatement("UPDATE temperatures SET temperature = ? WHERE sensor_id = ?")
  }

  //调用连接，执行sql
  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    //执行更新语句
    updateStat.setDouble(1, value.temperature)
    updateStat.setString(2, value.id)
    updateStat.execute()
    //如果更新没有该数据，则新增
    if (updateStat.getUpdateCount == 0){
      insertStat.setString(1, value.id)
      insertStat.setDouble(2, value.temperature)
      insertStat.execute()
    }
  }
  //关闭连接
  override def close(): Unit = {
    insertStat.close()
    updateStat.close()
    conn.close()
  }

}
